package com.xrui.hbase.impl;

import com.google.common.base.Preconditions;

import com.xrui.hbase.*;
import com.xrui.hbase.exception.ColumnPagingNotEnabledException;
import com.xrui.hbase.filter.ColumnFilter;
import com.xrui.hbase.filter.ColumnRangeFilter;
import com.xrui.hbase.filter.Filters;
import com.xrui.hbase.filter.StripValueColumnFilter;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.util.Debug;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicReference;

/**
 * HBase implementation of Pager for map-type families.
 * <p>
 * <p>
 * This pager lists the qualifiers in the map-type family and nothing else.
 * In particular, the cells content is not retrieved.
 * </p>
 * <p>
 * <p>
 * This pager conforms to the Pager interface, in order to implement
 * {@link RowData#getPager(String)}.
 */
public final class HBaseMapFamilyPager implements Pager {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseMapFamilyPager.class);

    /**
     * Entity ID of the row being paged through.
     */
    private final EntityId mEntityId;

    /**
     * HBaseTable to read from.
     */
    private final HBaseTable mTable;

    /**
     * Name of the map-type family being paged through.
     */
    private final ColumnName mFamily;

    /**
     * Full data request.
     */
    private final DataRequest mDataRequest;

    /**
     * Column data request for the map-type family to page through.
     */
    private final DataRequest.Column mColumnRequest;
    /**
     * Tracks the state of this map family pager.
     */
    private final AtomicReference<State> mState = new AtomicReference<State>(State.UNINITIALIZED);
    /**
     * True only if there is another page of data to read through {@link #next()}.
     */
    private boolean mHasNext;
    /**
     * Highest qualifier (according to the HBase bytes comparator) returned so far.
     * This is the low bound (exclusive) for qualifiers to retrieve next.
     */
    private String mMinQualifier = null;

    /**
     * Initializes a pager for a map-type family.
     * <p>
     * <p>
     * To create a pager for a column with paging enabled, use {@link RowData#getPager(String)}.
     * </p>
     *
     * @param entityId    The entityId of the row.
     * @param dataRequest The requested data.
     * @param table       The table that this row belongs to.
     * @param family      Iterate through the qualifiers from this map-type family.
     * @throws ColumnPagingNotEnabledException If paging is not enabled for the specified family.
     */
    HBaseMapFamilyPager(
        EntityId entityId,
        DataRequest dataRequest,
        HBaseTable table,
        ColumnName family)
        throws ColumnPagingNotEnabledException {

        Preconditions.checkArgument(!family.isFullyQualified(),
            "Must use HBaseQualifierPager on a map-type family, but got '{}'.", family);
        mFamily = family;

        mDataRequest = dataRequest;
        mColumnRequest = mDataRequest.getColumn(family.getFamily(), null);
        if (!mColumnRequest.isPagingEnabled()) {
            throw new ColumnPagingNotEnabledException(
                String.format("Paging is not enabled for column [%s].", family));
        }

        mEntityId = entityId;
        mTable = table;
        mHasNext = true;  // there might be no page to read, but we don't know until we issue an RPC

        // Only retain the table if everything else ran fine:
        mTable.retain();

        final State oldState = mState.getAndSet(State.OPEN);
        Preconditions.checkState(oldState == State.UNINITIALIZED,
            "Cannot open MapFamilyPager instance in state %s.", oldState);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean hasNext() {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot check has next while MapFamilyPager is in state %s.", state);
        return mHasNext;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public RowData next() {
        return next(mColumnRequest.getPageSize());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public RowData next(int pageSize) {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot create next while MapFamilyPager is in state %s.", state);
        if (!mHasNext) {
            throw new NoSuchElementException();
        }
        Preconditions.checkArgument(pageSize > 0, "Page size must be >= 1, got %s", pageSize);

        // Clone the column data request template, adjusting the filters to restrict the range
        // and the number of qualifiers to fetch.

        final ColumnFilter filter = Filters.and(
            new ColumnRangeFilter(mMinQualifier, false, null, false),  // qualifier > mMinQualifier
            mColumnRequest.getFilter(),  // user filter
            new PaginationFilter(pageSize),  // Select at most one version / qualifier.
            new StripValueColumnFilter());  // discard the cell content, we just need the qualifiers

        final DataRequest nextPageDataRequest = DataRequest.builder()
            .withTimeRange(mDataRequest.getMinTimestamp(), mDataRequest.getMaxTimestamp())
            .addColumns(DataRequestBuilder.ColumnsDef.create()
                .withFilter(filter)
                .withMaxVersions(1)  // HBase pagination filter forces max-versions to 1
                .add(mFamily))
            .build();

        LOG.debug("HBaseMapPager data request: {} and page size {}", nextPageDataRequest, pageSize);

        final TableLayout layout = mTable.getLayout();
        final HBaseDataRequestAdapter adapter =
            new HBaseDataRequestAdapter(nextPageDataRequest);
        try {
            final Get hbaseGet = adapter.toGet(mEntityId, layout);
            if (LOG.isDebugEnabled()) {
                LOG.debug("Sending HBase Get: {} with filter {}",
                    hbaseGet, Debug.toDebugString(hbaseGet.getFilter()));
            }
            final Result result = doHBaseGet(hbaseGet);
            LOG.debug("Got {} cells over {} requested", result.size(), pageSize);

            final RowData page =
                // No cell is being decoded here so we don't need a cell decoder provider:
                new HBaseRowData(mTable, nextPageDataRequest, mEntityId, result, null);

            // There is an HBase bug that leads to less KeyValue being returned than expected.
            // An empty result appears to be a reliable way to detect the end of the iteration.
            if (result.isEmpty()) {
                mHasNext = false;
            } else {
                // Update the low qualifier bound for the next iteration:
                mMinQualifier = page.getQualifiers(mFamily.getFamily()).last();
            }

            return page;

        } catch (IOException ioe) {
            throw new com.xrui.hbase.exception.IOException(ioe);
        }
    }

    /**
     * Sends an HBase Get request.
     *
     * @param get HBase Get request.
     * @return the HBase Result.
     * @throws IOException on I/O error.
     */
    private Result doHBaseGet(Get get) throws IOException {
        final HTableInterface htable = mTable.openHTableConnection();
        try {
            return htable.get(get);
        } finally {
            htable.close();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void remove() {
        throw new UnsupportedOperationException("Pager.remove() is not supported.");
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void close() throws IOException {
        final State oldState = mState.getAndSet(State.CLOSED);
        Preconditions.checkState(oldState == State.OPEN,
            "Cannot close MapFamilyPager while in state %s", oldState);
        mTable.release();
    }

    /**
     * The state of a map family pager instance.
     */
    private static enum State {
        UNINITIALIZED,
        OPEN,
        CLOSED
    }
}
